package cn.initcap;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * @author initcap
 * @date Created in 8/23/19 5:11 PM.
 */
public class BatchFlinkJava {

    public static void main(String[] args) throws Exception {
        String input = "file:///Users/ningzhixue/Public/code/initcap-learn/initcap-flink-core/input";

        // 1. 获取运行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 2. read data
        DataSource<String> text = env.readTextFile(input);

        // 3.transform
        text.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] tokens = s.toLowerCase().split(",");
                for (String token : tokens) {
                    if (token.length() > 0) {
                        collector.collect(new Tuple2<>(token, 1));
                    }
                }
            }
        }).groupBy(0).sum(1).print();
    }

}
